package com.atguigu.edu.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.base.BaseApp;
import com.atguigu.edu.realtime.common.base.BeanToJsonStrMapFunction;
import com.atguigu.edu.realtime.common.bean.TradePaperScoreStatistics;
import com.atguigu.edu.realtime.common.bean.TradeQuestionStatistics;
import com.atguigu.edu.realtime.common.util.DateFormatUtil;
import com.atguigu.edu.realtime.common.util.FlinkSinkUtil;
import com.atguigu.edu.realtime.common.util.HBaseUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.client.Connection;

public class DwsExamineQuestionStatistic extends BaseApp {
    public static void main(String[] args) {
        new DwsExamineQuestionStatistic().start(
                1498,
                4,
                "dws_exam_question_statistics_window",
                "dwd_exam_question"
        );
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String jsonStr, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                        if (StringUtils.isNotEmpty(jsonStr)) {
                            JSONObject jsonObj = JSON.parseObject(jsonStr);
                            collector.collect(jsonObj);
                        }
                    }
                }
        );

        KeyedStream<JSONObject, String> questionIdDS
                = jsonObjDS.keyBy(jsonObj -> jsonObj.getString("question_id"));

        //questionIdDS.print();

        SingleOutputStreamOperator<JSONObject> questionKeyedDS = questionIdDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<JSONObject>() {
                                    @Override
                                    public long extractTimestamp(JSONObject jsonObject, long l) {
                                        return jsonObject.getLong("ts")*1000;
                                    }
                                }
                        )
        );
        SingleOutputStreamOperator<TradeQuestionStatistics> beanDS = questionIdDS.process(
                new KeyedProcessFunction<String, JSONObject, TradeQuestionStatistics>() {
                    MapState<String, Integer> userMap;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        MapStateDescriptor<String, Integer> userMapDesc = new MapStateDescriptor<>("userMapDesc", String.class, Integer.class);
                        userMapDesc.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                        userMap = getRuntimeContext().getMapState(userMapDesc);
                    }

                    @Override
                    public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, TradeQuestionStatistics>.Context context, Collector<TradeQuestionStatistics> collector) throws Exception {
                        String userId = jsonObject.getString("user_id");
                        Long ts = jsonObject.getLong("ts")*1000;
                        String questionId = jsonObject.getString("question_id");
                        String isCorrect = jsonObject.getString("is_correct");
                        TradeQuestionStatistics orderBean = TradeQuestionStatistics.builder()
                                .questionId(questionId)
                                .yesQuesCt(0L)
                                .quesCt(1L)
                                .yesOneCt(0L)
                                .oneCt(0L)
                                .ts(ts)
                                .build();
                        if (isCorrect.equals("1")) {
                            orderBean.setYesQuesCt(1L);
                        }
                        if (userMap.get(userId) == null) {
                            orderBean.setOneCt(1L);
                        }
                        if (userMap.get(userId) == null && isCorrect.equals("1")) {
                            orderBean.setYesOneCt(1L);
                        }
                    collector.collect(orderBean);
                    }
                }
        );
        //beanDS.print();
        KeyedStream<TradeQuestionStatistics, String> keyedDS = beanDS.keyBy(TradeQuestionStatistics::getQuestionId);

        WindowedStream<TradeQuestionStatistics, String, TimeWindow> windowDS = keyedDS.window(TumblingProcessingTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));

        SingleOutputStreamOperator<TradeQuestionStatistics> reduceDS = windowDS.reduce(
                new ReduceFunction<TradeQuestionStatistics>() {
                    @Override
                    public TradeQuestionStatistics reduce(TradeQuestionStatistics v1, TradeQuestionStatistics v2) throws Exception {
                        v1.setYesQuesCt(v1.getYesQuesCt() + v2.getYesQuesCt());
                        v1.setQuesCt(v1.getQuesCt() + v2.getQuesCt());
                        v1.setYesOneCt(v1.getYesOneCt() + v2.getYesOneCt());
                        v1.setOneCt(v1.getOneCt() + v2.getOneCt());
                        return v1;
                    }
                },
                new WindowFunction<TradeQuestionStatistics, TradeQuestionStatistics, String, TimeWindow>() {
                    @Override
                    public void apply(String jsonStr, TimeWindow timeWindow, Iterable<TradeQuestionStatistics> iterable, Collector<TradeQuestionStatistics> collector) throws Exception {
                        TradeQuestionStatistics orderBean = iterable.iterator().next();
                        orderBean.setStt(DateFormatUtil.tsToDateTime(timeWindow.getStart()));
                        orderBean.setEdt(DateFormatUtil.tsToDateTime(timeWindow.getEnd()));
                        orderBean.setCurDate(DateFormatUtil.tsToDate(timeWindow.getStart()));
                        collector.collect(orderBean);
                    }
                }
        );
        //reduceDS.print();

        SingleOutputStreamOperator<TradeQuestionStatistics> realDS = reduceDS.map(
                new MapFunction<TradeQuestionStatistics, TradeQuestionStatistics>() {
                    @Override
                    public TradeQuestionStatistics map(TradeQuestionStatistics tradeQuestionStatistics) throws Exception {
                        Long yesQuesCt = tradeQuestionStatistics.getYesQuesCt();
                        Long quesCt = tradeQuestionStatistics.getQuesCt();
                        tradeQuestionStatistics.setAccuracy((double) yesQuesCt / quesCt);
                        Long yesOneCt = tradeQuestionStatistics.getYesOneCt();
                        Long oneCt = tradeQuestionStatistics.getOneCt();
                        tradeQuestionStatistics.setYesCtAllCt((double) yesOneCt / oneCt);
                        return tradeQuestionStatistics;
                    }
                }
        );

        SingleOutputStreamOperator<TradeQuestionStatistics> realstDS = realDS.map(
                new RichMapFunction<TradeQuestionStatistics, TradeQuestionStatistics>() {
                    Connection hBaseconn;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        hBaseconn = HBaseUtil.getHBaseConnection();
                    }

                    @Override
                    public void close() throws Exception {
                        HBaseUtil.closeHBaseConnection(hBaseconn);
                    }

                    @Override
                    public TradeQuestionStatistics map(TradeQuestionStatistics tradeQuestionStatistics) throws Exception {
                        String questionId = tradeQuestionStatistics.getQuestionId();
                        JSONObject dimBaseCourseInfo = HBaseUtil.getRow(hBaseconn, "edu", "dim_test_question_info", questionId, JSONObject.class);
                        tradeQuestionStatistics.setQuestionTxt(dimBaseCourseInfo.getString("question_txt"));

                        return tradeQuestionStatistics;
                    }
                }
        );
        realstDS.print();
        realstDS.map(new BeanToJsonStrMapFunction<>())
                .sinkTo(FlinkSinkUtil.getDorisSink("dws_exam_question_statistics_window"));
    }
}
